﻿#include "mqtt.h"
#include "platform.h"
//#pragma warning(disable: 6001) 
//#pragma warning(disable:26812) 
MQTT_List root_head;
MQTT_List* root = &root_head;
static int MQTT_Client_Name_Index=0;//MQTT句柄数量索引
/// <summary>
/// 获取输入报文的剩余长度，不包括固定报头
/// </summary>
/// <param name="InBuf">接收的报文</param>
/// <param name="remainLen">剩余长度</param>
/// <returns>剩余长度本身占据了几个字节</returns>
int Get_Remaining_Length(const char* InBuf, int* remainLen)
{
	const char* ptr = InBuf;
	int len = 0;
	int i = 1;
	int j = 0;
	ptr++;
	do {
		len += (*ptr & 0x7f) * i;
		i *= 128;
		j++;
		if (i > 128 * 128 * 128)
			return -1;
	} while (((*ptr++) & 0x80));
	*remainLen = len;
	return j;
}
/// <summary>
/// 将src长度为len的内容拷贝到ptr的地址上
/// </summary>
/// <param name="src">源地址</param>
/// <param name="dest">目的地址</param>
/// <param name="len">需拷贝的内容长度</param>
/// <returns></returns>
char* MQTT_Copy(char* src, char* dest, int len)
{
	if (dest == 0 || src == 0)
	{
		return 0;
	}
	memcpy(dest, src, len);
	return (dest + len);
}
/**
 * @brief Set_Remaining_Length 设置剩余长度
 * @param InBuf 报文剩余长度位置
 * @param len   剩余长度后续字节数
 * @return      设置了几个字节
 */
int Set_Remaining_Length(char* InBuf, int len)
{
	char remain = 0;
	int i = 0;
	do {
		remain = len % 128;
		len = len / 128;
		if (len > 0)
		{
			*InBuf++ = remain | 128;
		}
		else
			*InBuf = remain;
		i++;
	} while (len > 0);
	return i;
}
int Packet_CONNECT(MQTT* dev)
{
	int len = 0;
	char* ptr = dev->write_buf;
	char buf[10] = { 0 };
	uint16_t len_str = 0;
	uint16_t keep = 0;

	FixedHeader fixedHeader = { 0 };
	fixedHeader.head._Head.head_type = CONNECT;
	fixedHeader.head._Head.flags = 0;
	buf[0] = 0;
	buf[1] = 4;
	memcpy(&buf[2], "MQTT", strlen("MQTT"));
	buf[6] = 4;//协议级别
	if (dev->MQTT_Connect.connectflags._ConnectFlags.WillFlag == 0)
	{
		dev->MQTT_Connect.connectflags._ConnectFlags.WillQos = 0;
		dev->MQTT_Connect.connectflags._ConnectFlags.WillRetain = 0;
	}
	if (strlen(dev->UserName) == 0)
	{
		dev->MQTT_Connect.connectflags._ConnectFlags.PassWordFlag = 0;
		dev->MQTT_Connect.connectflags._ConnectFlags.UserNameFlag = 0;
	}
	else
	{
		dev->MQTT_Connect.connectflags._ConnectFlags.UserNameFlag = 1;
	}
	if (strlen(dev->PassWord) == 0)
	{
		dev->MQTT_Connect.connectflags._ConnectFlags.PassWordFlag = 0;
	}
	else
	{
		dev->MQTT_Connect.connectflags._ConnectFlags.PassWordFlag = 1;
	}
	dev->MQTT_Connect.connectflags._ConnectFlags.RESERVED = 0;
	if (strlen(dev->ClientID) == 0)
		dev->MQTT_Connect.connectflags._ConnectFlags.CleanSession = 0;
	buf[7] = dev->MQTT_Connect.connectflags.Flags;
	keep = honts(dev->MQTT_Connect.Keep_Alive);
	memcpy(&buf[8], &keep, sizeof(dev->MQTT_Connect.Keep_Alive));
	len = 10 + len + GET_STRING_LEN(dev->ClientID) +
		GET_STRING_LEN(dev->MQTT_Connect.Will_Topic) + GET_STRING_LEN(dev->MQTT_Connect.Will_Message) +
		GET_STRING_LEN(dev->UserName) + GET_STRING_LEN(dev->PassWord);
	fixedHeader.Remainlen = Set_Remaining_Length((char*)fixedHeader.Remaining_Length, (int)len);
	if (fixedHeader.Remainlen < 0)
		return -1;
	*ptr++ = fixedHeader.head.Flags;
	memcpy(ptr, fixedHeader.Remaining_Length, fixedHeader.Remainlen);
	ptr += fixedHeader.Remainlen;
	memcpy(ptr, buf, sizeof(buf));
	ptr += sizeof(buf);
	if (strlen(dev->ClientID) != 0)
	{
		len_str = honts(strlen(dev->ClientID));
		memcpy(ptr, &len_str, 2);
		ptr += 2;

		memcpy(ptr, &dev->ClientID, strlen(dev->ClientID));
		ptr += strlen(dev->ClientID);
	}

	if (dev->MQTT_Connect.connectflags._ConnectFlags.WillFlag != 0)
	{
		len_str = honts(strlen(dev->MQTT_Connect.Will_Topic));
		memcpy(ptr, &len_str, 2);
		ptr += 2;
		memcpy(ptr, &dev->MQTT_Connect.Will_Topic, strlen(dev->MQTT_Connect.Will_Topic));
		ptr += strlen(dev->MQTT_Connect.Will_Topic);

		len_str = honts(strlen(dev->MQTT_Connect.Will_Message));
		memcpy(ptr, &len_str, 2);
		ptr += 2;
		memcpy(ptr, &dev->MQTT_Connect.Will_Message, strlen(dev->MQTT_Connect.Will_Message));
		ptr += strlen(dev->MQTT_Connect.Will_Message);
	}
	if (strlen(dev->UserName) != 0)
	{
		len_str = honts(strlen(dev->UserName));
		memcpy(ptr, &len_str, 2);
		ptr += 2;
		memcpy(ptr, &dev->UserName, strlen(dev->UserName));
		ptr += strlen(dev->UserName);
	}
	if (strlen(dev->PassWord) != 0)
	{
		len_str = honts(strlen(dev->PassWord));
		memcpy(ptr, &len_str, 2);
		ptr += 2;
		memcpy(ptr, &dev->PassWord, strlen(dev->PassWord));
		ptr += strlen(dev->PassWord);
	}
	dev->Current_State = CONNECT;
	dev->write_len = (int)len + fixedHeader.Remainlen + 1;
	return 0;
}
int Unpack_CONNECT(MQTT* dev)
{
	int len = 0;
	char* ptr = dev->read_buf;
	char buf[5] = { 0 };
	uint16_t len_str = 0;
	uint16_t keep = 0;
	int i = 0;
	FixedHeader* fixedHeader = (FixedHeader*)ptr;
	if (fixedHeader->head._Head.head_type != CONNECT)
	{
		DEBUG_RAW("协议类型不正确", dev->read_buf, dev->read_len, 0);
		return -1;
	}
	if (fixedHeader->head._Head.flags != 0)
	{
		DEBUG_RAW("协议标志不正确", dev->read_buf, dev->read_len, 0);
		return -1;
	}
	ptr += Get_Remaining_Length(ptr, &len) + 3;
	memcpy(&buf[0], ptr, strlen("MQTT"));
	if (strcmp(buf, "MQTT") != 0)
	{
		DEBUG_RAW("协议名字mqtt不正确", dev->read_buf, dev->read_len, 0);
		return -1;
	}
	if (*(ptr + 4) != 4)
	{
		DEBUG_RAW("协议级别不正确", dev->read_buf, dev->read_len, 0);
		return -1;
	}
	ptr += 5;

	memcpy(&dev->MQTT_Connect.connectflags, ptr, 1);
	ptr++;

	memcpy(&dev->MQTT_Connect.Keep_Alive, ptr, 2);
	dev->MQTT_Connect.Keep_Alive = honts(dev->MQTT_Connect.Keep_Alive);
	ptr += 2;
	memcpy(dev->ClientID, ptr + 2, UINT8TOUINT16(ptr));
	ptr += 2 + UINT8TOUINT16(ptr);

	if (dev->MQTT_Connect.connectflags._ConnectFlags.WillFlag != 0)
	{
		memcpy(dev->MQTT_Connect.Will_Topic, ptr + 2, UINT8TOUINT16(ptr));
		ptr += 2 + UINT8TOUINT16(ptr);
		memcpy(dev->MQTT_Connect.Will_Message, ptr + 2, UINT8TOUINT16(ptr));
		ptr += 2 + UINT8TOUINT16(ptr);
	}
	if (dev->MQTT_Connect.connectflags._ConnectFlags.UserNameFlag != 0)
	{
		memcpy(dev->UserName, ptr + 2, UINT8TOUINT16(ptr));
		ptr += 2 + UINT8TOUINT16(ptr);
	}
	if (dev->MQTT_Connect.connectflags._ConnectFlags.PassWordFlag != 0)
	{
		memcpy(dev->PassWord, ptr + 2, UINT8TOUINT16(ptr));
		ptr += 2 + UINT8TOUINT16(ptr);
	}
	dev->Current_State = CONNECT;
	DEBUG_STRING("CONNECT,ClientID", dev->ClientID);
	if (dev->MQTT_Connect.connectflags._ConnectFlags.WillFlag != 0)
	{
		DEBUG_STRING("CONNECT,Will_Topic", dev->MQTT_Connect.Will_Topic);
		DEBUG_STRING("CONNECT,Will_Message", dev->MQTT_Connect.Will_Message);
	}
	if (dev->MQTT_Connect.connectflags._ConnectFlags.UserNameFlag != 0)
	{
		DEBUG_STRING("CONNECT,UserName", dev->UserName);
	}
	if (dev->MQTT_Connect.connectflags._ConnectFlags.PassWordFlag != 0)
	{
		DEBUG_STRING("CONNECT,PassWord", dev->PassWord);
	}
	
	return 0;
}
int UnPack_CONNACK(MQTT* dev)
{
	if (dev->read_len > 4)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != CONNACK)
		return -1;
	if (fixedHead->Remaining_Length[0] != 2)
		return -1;
	dev->MQTT_Connect.currentCleanSession = dev->read_buf[3];
	dev->MQTT_Connect.Connack_ErrCode = dev->read_buf[4];
	dev->Current_State = CONNACK;
	dev->write_len = 0;
	DEBUG_RAW("CONNACK连接结果", dev->read_buf, dev->read_len, dev->MQTT_Connect.Connack_ErrCode);
	if (dev->MQTT_Connect.Connack_ErrCode == 0)
	{
		dev->MQTT_Connect.connect_state = CONNECT_SUCCESS;
		return 0;
	}
	return -1;
}
int Packet_CONNACK(MQTT* dev)
{
	char* ptr = dev->write_buf;
	FixedHeader fixedHead = { 0 };
	fixedHead.head._Head.head_type = CONNACK;
	fixedHead.Remaining_Length[0] = 2;
	memcpy(ptr, &fixedHead, 2);
	ptr += 2;
	if (dev->MQTT_Connect.currentCleanSession == 1)
	{
		*ptr++ = 0;//当前会话设置（Session Present）标志
		*ptr++ = 0;//返回码
	}
	else
	{
		*ptr++ = 0;//当前会话设置（Session Present）标志
		*ptr++ = 0;//返回码
	}
	dev->MQTT_Connect.Connack_ErrCode = 0;
	dev->write_len = 4;
	return 0;
}
int Packet_PUBLISH(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 0;
	uint16_t PublishTpioc_len = 0;
	uint16_t msg_len = 0;
	uint16_t idendifiter = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PUBLISH;
	PublishFlags publishflag;
	publishflag._PublishFlags.DUP = dev->MQTT_Publish.Dup;
	publishflag._PublishFlags.QOS = dev->MQTT_Publish.Qos;
	publishflag._PublishFlags.RETAIN = dev->MQTT_Publish.RETAIN;
	fixedHead.head._Head.flags = publishflag.Flags;
	PublishTpioc_len = (uint16_t)strlen(dev->MQTT_Publish.PublishTpioc);
	len += PublishTpioc_len;//发布主题长度
	len += 2;//发布主题长度

	switch (dev->MQTT_Publish.Qos)
	{
	case 0:

		break;
	case 1:
	case 2:
		len += 2;//报文标识符
		break;
	}
	len += dev->MQTT_Publish.len;//消息内容长度
	fixedHead.Remainlen = Set_Remaining_Length((char*)fixedHead.Remaining_Length, len);
	if (fixedHead.Remainlen < 0)
		return -1;
	memcpy(ptr, &fixedHead, (int)(1 + fixedHead.Remainlen));
	ptr += (int)1 + (int)fixedHead.Remainlen;


	PublishTpioc_len = honts(PublishTpioc_len);
	memcpy(ptr, &PublishTpioc_len, 2);
	ptr += 2;
	memcpy(ptr, &dev->MQTT_Publish.PublishTpioc, honts(PublishTpioc_len));
	ptr += honts(PublishTpioc_len);
	switch (dev->MQTT_Publish.Qos)
	{
	case 0:

		break;
	case 1:
	case 2:
		idendifiter = honts(dev->MQTT_Publish.Packet_Identifier);
		memcpy(ptr, &idendifiter, sizeof(dev->MQTT_Publish.Packet_Identifier));
		ptr += sizeof(dev->MQTT_Publish.Packet_Identifier);
		break;
	}
	if(dev->MQTT_Publish.len!=0)
		memcpy(ptr, dev->MQTT_Publish.message, dev->MQTT_Publish.len);
	dev->Current_State = PUBLISH;
	dev->write_len = len + fixedHead.Remainlen + 1;
	return 0;
}
int Unpack_PUBLISH(MQTT* dev)
{
	int len = 0;
	int i = 0;
	uint16_t len_tpioc = 0;
	char* ptr = dev->read_buf, *ploay=0;
	PublishFlags publishflag = { 0 };
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != PUBLISH)
		return -1;

	i = Get_Remaining_Length(ptr, &len);
	if (dev->read_len != len + i + 1)
		return -1;
	ptr += (int)(i + 1);

	publishflag.Flags = fixedHead->head._Head.flags;
	dev->MQTT_Unpublish.UnPublishQos = publishflag._PublishFlags.QOS;
	len_tpioc = *ptr << 8 | (*(ptr + 1));//下发的消息主题长度
	ptr += 2;
	memcpy(dev->MQTT_Unpublish.UnPublishTpioc, ptr, len_tpioc);
	ptr += len_tpioc;
	if (dev->MQTT_Unpublish.UnPublishQos != 0)
	{
		memcpy(&dev->MQTT_Unpublish.UnPublish_Identifier, ptr, 2);
		dev->MQTT_Unpublish.UnPublish_Identifier = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
		ptr += 2;
		dev->MQTT_Unpublish.PublishPloay_len = len - len_tpioc - 4;
	}
	else
		dev->MQTT_Unpublish.PublishPloay_len = len - len_tpioc - 2;
	if (dev->MQTT_Unpublish.PublishPloay_len != 0)
	{
		MQTT_POLAY_COPY(ploay, dev->MQTT_Unpublish.PublishPloay, ptr, dev->MQTT_Unpublish.PublishPloay_len);

		if (ploay == NULL)
			return -1;
	}
	
	dev->Current_State = PUBLISH;
	DEBUG_STRING("收到PUBLISH报文 主题", dev->MQTT_Unpublish.UnPublishTpioc);
	if (dev->MQTT_Unpublish.PublishPloay_len != 0)
	{
		DEBUG_STRING("收到PUBLISH报文 消息", dev->MQTT_Unpublish.PublishPloay);
	}
	else
	{
		DEBUG_STRING("收到PUBLISH报文 消息", "空");
	}
	return 0;
}
int UnPack_PUBACK(MQTT* dev)
{
	if (dev->read_len > 4)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != PUBACK)
		return -1;
	if (fixedHead->Remaining_Length[0] != 2)
		return -1;
	memcpy(&dev->MQTT_Unpublish.UnPublish_Identifier, &dev->read_buf[2], 2);
	dev->Current_State = PUBACK;
	dev->write_len = 0;
	DEBUG_RAW("QOS=1 收到PUBACK", dev->read_buf, dev->read_len, dev->Current_State);
	return 0;
}
int Packet_PUBACK(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PUBACK;
	fixedHead.head._Head.flags = 0;
	fixedHead.Remaining_Length[0] = 2;

	memcpy(ptr, &fixedHead.head, 2);
	ptr += 2;
	uint16_t id = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	memcpy(ptr, &id, 2);
	dev->Current_State = PUBACK;
	len += 4;
	dev->write_len = len;
	return 0;
}
int UnPack_PUBREC(MQTT* dev)
{
	if (dev->read_len > 4)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != PUBREC)
		return -1;
	if (fixedHead->Remaining_Length[0] != 2)
		return -1;
	memcpy(&dev->MQTT_Unpublish.UnPublish_Identifier, &dev->read_buf[2], 2);
	dev->MQTT_Unpublish.UnPublish_Identifier = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	dev->Current_State = PUBREC;
	DEBUG_RAW("QOS=2，收到PUBREC", dev->read_buf, dev->read_len, dev->Current_State);
	return 0;
}
int Packet_PUBREC(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 2;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PUBREC;
	fixedHead.head._Head.flags = 0;
	fixedHead.Remaining_Length[0] = 2;

	memcpy(ptr, &fixedHead.head, 2);
	ptr += 2;
	uint16_t id = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	memcpy(ptr, &id, 2);
	dev->Current_State = PUBREC;
	dev->write_len = 4;
	return 0;
}
int UnPack_PUBREL(MQTT* dev)
{
	if (dev->read_len > 4)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != PUBREL)
		return -1;
	if (fixedHead->Remaining_Length[0] != 2)
		return -1;
	memcpy(&dev->MQTT_Unpublish.UnPublish_Identifier, &dev->read_buf[2], 2);
	dev->MQTT_Unpublish.UnPublish_Identifier = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	dev->Current_State = PUBREL;
	return 1;
}
int Packet_PUBREL(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 2;
	uint16_t idendifiter = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PUBREL;
	fixedHead.head._Head.flags = 2;
	fixedHead.Remaining_Length[0] = 2;

	memcpy(ptr, &fixedHead.head, 2);
	ptr += 2;
	idendifiter = honts(dev->MQTT_Publish.Packet_Identifier);
	memcpy(ptr, &idendifiter, 2);
	dev->Current_State = PUBREL;
	dev->write_len = 4;
	return 0;
}
int UnPack_PUBCOMP(MQTT* dev)
{
	if (dev->read_len > 4)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != PUBCOMP)
		return -1;
	if (fixedHead->Remaining_Length[0] != 2)
		return -1;
	memcpy(&dev->MQTT_Unpublish.UnPublish_Identifier, &dev->read_buf[2], 2);
	dev->MQTT_Unpublish.UnPublish_Identifier = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	dev->Current_State = PUBCOMP;
	DEBUG_RAW("QOS=2，收到PUBCOMP,发送结束", dev->read_buf, dev->read_len, dev->Current_State);
	return 1;
}
int Packet_PUBCOMP(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 2;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PUBCOMP;
	fixedHead.head._Head.flags = 0;
	fixedHead.Remaining_Length[0] = 2;

	memcpy(ptr, &fixedHead.head, 2);
	ptr += 2;
	uint16_t id = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	memcpy(ptr, &id, 2);
	dev->Current_State = PUBCOMP;
	dev->write_len = 4;
	return 0;
}
int Packet_SUBSCRIBE(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 0;
	uint16_t tpioc_len = 0;
	uint16_t idendifiter = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = SUBSCRIBE;
	fixedHead.head._Head.flags = 2;
	tpioc_len = (uint16_t)strlen(dev->MQTT_Subscribe.SubscribeTpioc);
	len += 2 + tpioc_len + 2 + 1;
	fixedHead.Remainlen = Set_Remaining_Length((char*)fixedHead.Remaining_Length, len);
	if (fixedHead.Remainlen < 0)
		return -1;
	memcpy(ptr, &fixedHead, (int)1 + fixedHead.Remainlen);
	ptr += (int)1 + fixedHead.Remainlen;
	idendifiter = honts(dev->MQTT_Publish.Packet_Identifier);
	memcpy(ptr, &idendifiter, sizeof(dev->MQTT_Publish.Packet_Identifier));
	ptr += sizeof(dev->MQTT_Publish.Packet_Identifier);
	tpioc_len = honts(tpioc_len);
	memcpy(ptr, &tpioc_len, sizeof(tpioc_len));
	ptr += sizeof(tpioc_len);
	memcpy(ptr, dev->MQTT_Subscribe.SubscribeTpioc, strlen(dev->MQTT_Subscribe.SubscribeTpioc));
	ptr += strlen(dev->MQTT_Subscribe.SubscribeTpioc);
	*ptr = dev->MQTT_Subscribe.Qos;
	dev->Current_State = SUBSCRIBE;
	dev->write_len = len + fixedHead.Remainlen + 1;
	return len + fixedHead.Remainlen + 1;
}
int UnPack_SUBSCRIBE(MQTT* dev)
{
	FixedHeader* fixedHead = 0;
	int len = 0;
	uint16_t tpioc_len = 0;
	uint16_t idendifiter = 0;
	char* ptr = dev->read_buf;
	fixedHead = (FixedHeader*)ptr;

	if (fixedHead->head._Head.head_type != SUBSCRIBE)
		return -1;
	if (fixedHead->head._Head.flags != 2)
		return -1;
	ptr += Get_Remaining_Length(ptr, &len) + 1;
	memcpy(&dev->MQTT_Subscribe.SUBSCRIBE_Identifier, ptr, 2);
	dev->MQTT_Subscribe.SUBSCRIBE_Identifier = honts(dev->MQTT_Subscribe.SUBSCRIBE_Identifier);

	ptr += 2;
	memcpy(dev->MQTT_Subscribe.SubscribeTpioc, ptr + 2, UINT8TOUINT16(ptr));
	ptr += 2 + UINT8TOUINT16(ptr);
	dev->MQTT_Publish.Qos = *ptr;
	dev->Current_State = SUBSCRIBE;
	DEBUG_STRING("收到SUBSCRIBE主题", dev->MQTT_Subscribe.SubscribeTpioc);
	DEBUG_NUMBER("收到SUBSCRIBE标识符", dev->MQTT_Subscribe.SUBSCRIBE_Identifier);
	DEBUG_NUMBER("收到SUBSCRIBE QOS", dev->MQTT_Publish.Qos);
	return 0;
}
int UnPack_SUBACK(MQTT* dev)
{
	if (dev->read_len > 5)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != SUBACK)
		return -1;
	memcpy(&dev->MQTT_Unpublish.UnPublish_Identifier, &dev->read_buf[2], 2);
	dev->MQTT_Unpublish.UnPublish_Identifier = honts(dev->MQTT_Unpublish.UnPublish_Identifier);
	dev->MQTT_Subscribe.Subscribe_Result = dev->read_buf[4];
	dev->Current_State = SUBACK;
	DEBUG_RAW("订阅结果", dev->read_buf, dev->read_len, dev->MQTT_Subscribe.Subscribe_Result);
	return 1;
}
int Packet_SUBACK(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	uint16_t id = 0;
	int len = 1;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = SUBACK;
	fixedHead.head._Head.flags = 0;
	fixedHead.Remainlen += Set_Remaining_Length((char*)fixedHead.Remaining_Length, 3);
	memcpy(ptr, &fixedHead, 1 + fixedHead.Remainlen);
	ptr += 1 + fixedHead.Remainlen;
	len += 1 + fixedHead.Remainlen;
	id = honts(dev->MQTT_Subscribe.SUBSCRIBE_Identifier);
	memcpy(ptr, &id, 2);
	len += 2;
	ptr += 2;
	*ptr = 0;//订阅结果
	dev->Current_State = SUBACK;
	dev->write_len = len;
	return 0;
}
int Packet_UNSUBSCRIBE(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 0;
	int tpioc_len = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = UNSUBSCRIBE;
	fixedHead.head._Head.flags = 2;
	tpioc_len = (int)strlen(dev->MQTT_Subscribe.SubscribeTpioc);
	len += 2 + tpioc_len + 2;
	fixedHead.Remainlen = Set_Remaining_Length((char*)fixedHead.Remaining_Length, len);
	if (fixedHead.Remainlen < 0)
		return -1;
	memcpy(ptr, &fixedHead, (int)1 + (int)fixedHead.Remainlen);
	ptr += (int)1 + (int)fixedHead.Remainlen;
	memcpy(ptr, &dev->MQTT_Publish.Packet_Identifier, sizeof(dev->MQTT_Publish.Packet_Identifier));
	ptr += sizeof(dev->MQTT_Publish.Packet_Identifier);
	memcpy(ptr, &tpioc_len, sizeof(tpioc_len));
	ptr += sizeof(tpioc_len);
	memcpy(ptr, dev->MQTT_Subscribe.SubscribeTpioc, strlen(dev->MQTT_Subscribe.SubscribeTpioc));
	ptr += strlen(dev->MQTT_Subscribe.SubscribeTpioc);
	dev->Current_State = UNSUBSCRIBE;
	dev->write_len = len + fixedHead.Remainlen + 1;
	return len + fixedHead.Remainlen + 1;
}
int UnPack_UNSUBSCRIBE(MQTT* dev)
{
	FixedHeader* fixedHead = 0;
	int len = 0;
	int tpioc_len = 0;
	char* ptr = dev->read_buf;
	fixedHead = (FixedHeader*)ptr;

	if (fixedHead->head._Head.head_type != UNSUBSCRIBE)
		return -1;
	if (fixedHead->head._Head.flags != 2)
		return -1;
	ptr += Get_Remaining_Length(ptr, &len) + 1;
	memcpy(&dev->MQTT_Subscribe.SUBSCRIBE_Identifier, ptr, 2);
	dev->MQTT_Subscribe.SUBSCRIBE_Identifier = honts(dev->MQTT_Subscribe.SUBSCRIBE_Identifier);
	ptr += 2;
	memcpy(dev->MQTT_Subscribe.SubscribeTpioc, ptr + 2, UINT8TOUINT16(ptr));
	dev->Current_State = UNSUBSCRIBE;
	DEBUG_STRING("取消订阅主题", dev->MQTT_Subscribe.SubscribeTpioc);
	return 0;
}
int UnPack_UNSUBACK(MQTT* dev)
{
	if (dev->read_len > 4)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != UNSUBACK)
		return -1;
	if (fixedHead->Remainlen != 2)
		return -1;
	memcpy(&dev->MQTT_Subscribe.SUBSCRIBE_Identifier, &dev->read_buf[2], 2);
	dev->Current_State = UNSUBACK;
	return 1;
}
int Packet_UNSUBACK(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 0;
	uint16_t id = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = UNSUBACK;
	fixedHead.head._Head.flags = 0;
	fixedHead.Remaining_Length[0] = 2;
	memcpy(ptr, &fixedHead, 2);
	ptr += 2;
	id = honts(dev->MQTT_Subscribe.SUBSCRIBE_Identifier);
	memcpy(ptr, &id, 2);
	ptr += 2;
	len += 2 + 2;
	*ptr = 0;//订阅结果
	dev->Current_State = UNSUBACK;
	dev->write_len = len;
	return 0;
}
int Packet_PINGREQ(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 2;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PINGREQ;
	fixedHead.head._Head.flags = 0;

	memcpy(ptr, &fixedHead.head, 1);
	ptr += 1;
	dev->Current_State = PINGREQ;
	dev->write_len = len;
	return len;
}
int UnPack_PINGREQ(MQTT* dev)
{
	FixedHeader* fixedHead = 0;
	if (dev->read_len != 2)
		return -1;
	char* ptr = dev->read_buf;
	fixedHead = (FixedHeader*)ptr;
	if (fixedHead->head._Head.head_type != PINGREQ)
		return -1;
	if (fixedHead->head._Head.flags != 0)
		return -1;
	dev->Current_State = PINGREQ;
	return 0;
}
int UnPack_PINGRESP(MQTT* dev)
{
	if (dev->read_len > 2)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != PINGRESP)
		return -1;
	if (fixedHead->head._Head.flags != 0)
		return -1;
	if (fixedHead->Remaining_Length[0] != 0)
		return -1;
	dev->Current_State = PINGRESP;
	return 1;
}
int Packet_PINGRESP(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 0;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = PINGRESP;
	fixedHead.head._Head.flags = 0;
	fixedHead.Remainlen = 0;
	memcpy(ptr, &fixedHead, (int)fixedHead.Remainlen + 1);
	ptr += (int)fixedHead.Remainlen + 1;
	*ptr = 0;
	len += fixedHead.Remainlen + 1 + 1;
	dev->Current_State = PINGRESP;
	dev->write_len = len;
	return 0;
}
int Packet_DISCONNECT(MQTT* dev)
{
	FixedHeader fixedHead = { 0 };
	int len = 2;
	char* ptr = dev->write_buf;
	fixedHead.head._Head.head_type = DISCONNECT;
	fixedHead.head._Head.flags = 0;
	memcpy(ptr, &fixedHead.head, 1);
	dev->Current_State = DISCONNECT;
	return len;
}
int UnPack_DISCONNECT(MQTT* dev)
{
	if (dev->read_len > 2)
		return -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;
	if (fixedHead->head._Head.head_type != DISCONNECT)
		return -1;
	if (fixedHead->head._Head.flags != 0)
		return -1;
	if (fixedHead->Remaining_Length[0] != 0)
		return -1;
	dev->Current_State = DISCONNECT;
	DEBUG_RAW("客户端断开连接，请重新连接", dev->read_buf, dev->read_len, dev->Current_State);
	return 1;
}
int MQTT_RecvPoll(MQTT* dev)
{
	int result = -1;
	FixedHeader* fixedHead = (FixedHeader*)dev->read_buf;

	DEBUG_STRING("--------------------------------------------------------------------", " ");
	DEBUG_RAW("收到信息", dev->read_buf, dev->read_len, fixedHead->head._Head.head_type);
	switch (fixedHead->head._Head.head_type)
	{
	case RESERVED:
		break;
	case  CONNECT:
		result = Unpack_CONNECT(dev);
		if (result != -1)
		{
			result = Packet_CONNACK(dev);
			if (result != -1)
				result = dev->send_handle(dev);
			if (result != -1)
			{
				DEBUG_RAW("客户端CONNECT连接成功", dev->write_buf, dev->write_len, dev->Current_State);
				dev->MQTT_Connect.connect_state = CONNECT_SUCCESS;
			}
			else
			{
				DEBUG_RAW("客户端CONNECT连接失败", dev->write_buf, dev->write_len, dev->Current_State);
			}
		}
		break;
	case CONNACK:
		result = UnPack_CONNACK(dev);
		break;
	case PUBLISH:
		result = Unpack_PUBLISH(dev);
		{
			if (dev->mqtt_socket_type == MQTT_SERVER && result != -1)
			{
				switch (dev->MQTT_Unpublish.UnPublishQos)
				{
				case 0:
					DEBUG_RAW("收到客户端Publish成功", dev->read_buf, dev->read_len, dev->Current_State);
					Check_Subscribe_State(dev);
					break;
				case 1:
					result = Packet_PUBACK(dev);
					if (result != -1)
						result = dev->send_handle(dev);
					if (result != -1)
					{
						DEBUG_RAW("向客户端发送PUBACK成功", dev->write_buf, dev->write_len, dev->Current_State);
						Check_Subscribe_State(dev);
					}
					else
					{
						DEBUG_RAW("向客户端发送PUBACK失败", dev->write_buf, dev->write_len, dev->Current_State);
					}
					break;
				case 2:
					result = Packet_PUBREC(dev);
					if (result != -1)
						result = dev->send_handle(dev);
					if (result != -1)
					{
						DEBUG_RAW("向客户端发送PUBREC成功", dev->write_buf, dev->write_len, dev->Current_State);
					}
					else
					{
						DEBUG_RAW("向客户端发送PUBREC失败", dev->write_buf, dev->write_len, dev->Current_State);
					}
					break;
				default:
					break;
				}
			}
		}
		break;
	case PUBACK:
		result = UnPack_PUBACK(dev);
		break;
	case PUBREC:
		result = UnPack_PUBREC(dev);
		break;
	case PUBREL:
		result = UnPack_PUBREL(dev);
		{
			if (dev->mqtt_socket_type == MQTT_SERVER && result != -1)
			{
				switch (dev->MQTT_Unpublish.UnPublishQos)
				{
				case 0:
					break;
				case 1:
					break;
				case 2:
					result = Packet_PUBCOMP(dev);
					if (result != -1)
						result = dev->send_handle(dev);
					if (result != -1)
					{
						DEBUG_RAW("向客户端发送PUBCOMP成功", dev->write_buf, dev->write_len, dev->Current_State);
						Check_Subscribe_State(dev);
					}
					else
					{
						DEBUG_RAW("向客户端发送PUBCOMP失败", dev->write_buf, dev->write_len, dev->Current_State);
					}
					break;
				default:
					break;
				}
			}
		}
		break;
	case PUBCOMP:
		result = UnPack_PUBCOMP(dev);
		break;
	case SUBSCRIBE:
	{
		result = UnPack_SUBSCRIBE(dev);
		if (dev->mqtt_socket_type == MQTT_SERVER && result != -1)
		{
			result = Packet_SUBACK(dev);
			if (result != -1)
				result = dev->send_handle(dev);
			if (result != -1)
			{
				DEBUG_RAW("向客户端发送SUBACK成功", dev->write_buf, dev->write_len, dev->Current_State);
			}
			else
			{
				DEBUG_RAW("向客户端发送SUBACK失败", dev->write_buf, dev->write_len, dev->Current_State);
			}
		}
	}
	break;
	case SUBACK:
		result = UnPack_SUBACK(dev);
		break;
	case UNSUBSCRIBE:
		result = UnPack_UNSUBSCRIBE(dev);
		if (dev->mqtt_socket_type == MQTT_SERVER && result != -1)
		{
			result = Packet_UNSUBACK(dev);
			if (result != -1)
				result = dev->send_handle(dev);
			if (result != -1)
			{
				DEBUG_RAW("向客户端发送UNSUBACK成功", dev->write_buf, dev->write_len, dev->Current_State);
			}
			else
			{
				DEBUG_RAW("向客户端发送UNSUBACK失败", dev->write_buf, dev->write_len, dev->Current_State);
			}
		}
		break;
	case UNSUBACK:
		result = UnPack_UNSUBACK(dev);
		break;
	case PINGREQ:
		result = UnPack_PINGREQ(dev);
		if (dev->mqtt_socket_type == MQTT_SERVER && result != -1)
		{
			result = Packet_PINGRESP(dev);
			if (result != -1)
				result = dev->send_handle(dev);
			if (result != -1)
			{
				DEBUG_RAW("向客户端发送PINGRESP成功", dev->write_buf, dev->write_len, dev->Current_State);
			}
			else
			{
				DEBUG_RAW("向客户端发送PINGRESP失败", dev->write_buf, dev->write_len, dev->Current_State);
			}
		}
		break;
	case PINGRESP:
		result = UnPack_PINGRESP(dev);
		break;
	case DISCONNECT:
		result = UnPack_DISCONNECT(dev);
		break;
	}
	if (result == -1)
	{
		DEBUG_RAW("接收报文错误", dev->read_buf, dev->read_len, dev->Current_State);
	}
	return result;
}
MQTT* Create_mqtt(MQTT_Socket_Type TYPE)
{
	MQTT* mqtt = NULL;
	mqtt = (MQTT*)malloc(sizeof(MQTT));
	if (mqtt != NULL)
	{
		memset(mqtt, 0, sizeof(MQTT));
		mqtt->MQTT_Publish.Packet_Identifier = 1;
		mqtt->MQTT_Connect.Keep_Alive = KEEP_ALIVE;
		mqtt->read_len = BUFSIZE;
		mqtt->write_len = 0;
		mqtt->recv_handle = MQTT_RecvPoll;
		mqtt->MQTT_Connect.Connack_ErrCode = 0xff;
		mqtt->mqtt_socket_type = TYPE;
		mqtt->read_buf = (char*)malloc(BUFSIZE);
		mqtt->write_buf = (char*)malloc(BUFSIZE);
		if (mqtt->read_buf == NULL || mqtt->write_buf == NULL)
		{
			free(mqtt);
			if (mqtt->read_buf != NULL)
				free(mqtt->read_buf);
			if (mqtt->write_buf != NULL)
				free(mqtt->write_buf);
			return 0;
		}
		else
		{
			sprintf(mqtt->name,"MQTT:%d", MQTT_Client_Name_Index);
			MQTT_Client_Name_Index++;
			DEBUG_STRING("创建MQTT句柄", mqtt->name);
			memset(mqtt->read_buf, 0, BUFSIZE);
			memset(mqtt->write_buf, 0, BUFSIZE);
		}
	}
	return mqtt;
}
void Delete_mqtt(MQTT* dev)
{
	if (dev->MQTT_Publish.message != NULL)
	{
		free(dev->MQTT_Publish.message);
		dev->MQTT_Publish.message = NULL;
	}
	if (dev->read_buf != NULL)
		free(dev->read_buf);
	if (dev->write_buf != NULL)
		free(dev->write_buf);
	if (dev->mqtt_child_thread_handle != NULL)
		CloseHandle(dev->mqtt_child_thread_handle);
	if (dev->MQTT_Publish.message != NULL)
		free(dev->MQTT_Publish.message);
	if (dev->MQTT_Unpublish.PublishPloay != NULL)
		free(dev->MQTT_Unpublish.PublishPloay);
	MQTT_Client_Name_Index--;
	DEBUG_STRING("销毁MQTT句柄", dev->name);
	
}
void MQTT_SetClientParameter(MQTT* dev, const char* UserName, const char* PassWord, const char* ClienID, int CleanSession)
{
	if (dev == 0)
		return;
	if (ClienID != 0)
		memcpy(dev->ClientID, ClienID, strlen(ClienID));
	if (PassWord != 0)
		memcpy(dev->PassWord, PassWord, strlen(PassWord));
	if (UserName != 0)
		memcpy(dev->UserName, UserName, strlen(UserName));
	dev->MQTT_Connect.connectflags._ConnectFlags.CleanSession = CleanSession;
}
int MQTT_Publish(MQTT* dev, const char* PublishTpioc, MQTT_MSG* msg)
{
	char* ptr = 0;
	int flag = 0;
	if (dev->MQTT_Connect.Connack_ErrCode != 0)
		return -1;
	int re = 0;
	if (msg!= NULL)
	{
		dev->MQTT_Publish.message = msg->message;
		dev->MQTT_Publish.len = msg->len;
		dev->MQTT_Publish.Dup = msg->Dup;
		dev->MQTT_Publish.Qos = msg->Qos;
		memcpy(dev->MQTT_Publish.PublishTpioc, PublishTpioc, strlen(PublishTpioc));
	}	
	Packet_PUBLISH(dev);
	if (dev->send_handle != NULL)
		re = dev->send_handle(dev);
	if (dev->Current_State == PUBLISH && dev->MQTT_Publish.Qos == 0 && re >= 0)
	{
		DEBUG_RAW("QOS=0发布消息成功", dev->write_buf, dev->write_len, dev->Current_State);
		re |= 0x80;
		goto exit;
	}
	else if (dev->Current_State == PUBLISH && dev->MQTT_Publish.Qos == 0)
	{
		DEBUG_RAW("QOS=0发布消息失败", dev->write_buf, dev->write_len, dev->Current_State);
		re = -1;
		goto exit;
	}
	do
  	{
		mqtt_sem_wait(dev);
		if (dev->data_err == 1)
		{
			re = -1;
			break;
		}
		if (dev->Current_State == PUBACK && dev->MQTT_Publish.Qos == 1)
		{
			re |= 0x80;
			break;
		}
		if (dev->Current_State == PUBREC && dev->MQTT_Publish.Qos == 2)
		{
			Packet_PUBREL(dev);
			if (dev->send_handle != NULL)
				re = dev->send_handle(dev);
			mqtt_sem_post(dev);
			flag = 1;
		}
		if (dev->Current_State == PUBCOMP && dev->MQTT_Publish.Qos == 2)
		{
			re |= 0x80;
			mqtt_sem_post(dev);
			flag = 1;
			break;
		}
	} while (1);
	if (flag == 0)
		mqtt_sem_post(dev);//需要释放信号量
exit:
	return re;
}
/// <summary>
/// 查找所有订阅的客户端中与本次接收的Publish主题相符的客户端，将SubToPublish_flag需要转发消息的标志置1
/// </summary>
/// <param name="dev">本次接收发布消息的客户端句柄</param>
void Check_Subscribe_State(List_type* List_node)
{
	char* ptr = 0;
	MQTT* Mqtt = 0;
	int len = List_Length(root);
	while (len--)
	{
		Mqtt = List_node_by_PublishTpoic(List_node->MQTT_Unpublish.UnPublishTpioc);
		if (Mqtt != 0)
		{
			Mqtt->MQTT_Publish.Dup = 0;
			Mqtt->MQTT_Publish.len = List_node->MQTT_Unpublish.PublishPloay_len;
			if (List_node->MQTT_Unpublish.PublishPloay_len != 0)
			{
				MQTT_POLAY_COPY(ptr, Mqtt->MQTT_Publish.message, List_node->MQTT_Unpublish.PublishPloay, Mqtt->MQTT_Publish.len);

				if (ptr == NULL)
					continue;
			}			
			memcpy(Mqtt->MQTT_Publish.PublishTpioc, List_node->MQTT_Unpublish.UnPublishTpioc, strlen(List_node->MQTT_Unpublish.UnPublishTpioc));
			Mqtt->MQTT_Publish.Packet_Identifier = Mqtt->MQTT_Unpublish.UnPublish_Identifier;
			Mqtt->MQTT_Publish.Qos = List_node->MQTT_Unpublish.UnPublishQos;
			Mqtt->MQTT_Unpublish.SubToPublish_flag = 1;
			mqtt_subscribe_sem_post();
		}
	}
}
void List_Init(void)
{
	root->node = 0;
	root->next_node = root;
	root->prev_node = root;
}
/// <summary>
/// 创建链表节点
/// </summary>
/// <param name="data_node">要挂载的元素</param>
/// <returns>返回创建的节点</returns>
MQTT_List* List_Create_node(List_type* data_node)
{
	MQTT_List* L = NULL;
	L = (MQTT_List*)malloc(sizeof(struct _MQTT_LIST));
	if (L == NULL)
		return NULL;
	memset(L, 0, sizeof(struct _MQTT_LIST));
	L->node = data_node;
	L->next_node = L;
	L->prev_node = L;
	return L;
}
/// <summary>
/// 删除链表某个节点
/// </summary>
/// <param name="Item">要删除的节点</param>
void List_Delete_node(MQTT_List* Item)
{
	if (Item == NULL)
		return;
	Item->prev_node->next_node = Item->next_node;
	Item->next_node->prev_node = Item->prev_node;	
	Item->node = NULL;
	Item->next_node = NULL;
	Item->prev_node = NULL;
	free(Item);
}
/// <summary>
/// 删除链表
/// </summary>
/// <param name="Item">任意一个节点</param>
void List_Destroy(MQTT_List* Item)
{
	MQTT_List* tmp;
	int len = 0;
	if (Item == NULL)
		return;
	len = List_Length(Item);
	while (len-- && Item != NULL)
	{
		tmp = Item->prev_node;
		List_Delete_node(Item);
		Item = tmp;
	}
}
/// <summary>
/// 后插法
/// </summary>
/// <param name="root">要插入节点在此节点后</param>
/// <param name="Item">要插入节点</param>
void List_Insert_back(MQTT_List* root, MQTT_List* Item)
{
	if (root == NULL || Item == NULL)
		return;
	Item->prev_node = root;
	root->next_node->prev_node = Item;
	Item->next_node = root->next_node;
	root->next_node = Item;
}
/// <summary>
/// 前插法
/// </summary>
/// <param name="root">要插入节点在此节点前</param>
/// <param name="Item">要插入节点</param>
void List_Insert_forward(MQTT_List* root, MQTT_List* Item)
{
	if (root == NULL || Item == NULL)
		return;
	root->prev_node->next_node = Item;
	Item->prev_node = root->prev_node;
	Item->next_node = root;
	root->prev_node = Item;
}
/// <summary>
/// 获取链表长度
/// </summary>
/// <param name="Item">任意节点</param>
/// <returns>长度</returns>
int List_Length(MQTT_List* Item)
{
	MQTT_List* tmp, * it = Item;
	int i = 0;
	if (it == NULL)
		return 0;
	while (it!=NULL&&Item != it->prev_node)
	{
		tmp = it->prev_node;
		i++;
		it = tmp;
	}
	return i + 1;
}
/// <summary>
/// 寻找链表中的元素与输入元素相同的节点，并删除此节点
/// </summary>
/// <param name="mqtt">输入要查询的元素</param>
void List_Delete_node_by_data(List_type* List_node)
{
	MQTT_List* tmp, * p = root;
	int len = 0;
	if (List_node == NULL)
		return;
	len = List_Length(root);
	while (len-- && p != NULL)
	{
		tmp = p->prev_node;
		if (p->node == NULL)
		{
			p = tmp;
			continue;
		}
		if (p->node == List_node)
		{
			List_Delete_node(p);
			return;
		}
		p = tmp;
	}
}
MQTT* List_node_by_SubToPublish(void)
{
	MQTT_List* tmp, * p = root;
	int len = 0;

	len = List_Length(root);
	while (len-- && p != NULL)
	{
		tmp = p->prev_node;
		if (p->node == NULL)
		{
			p = tmp;
			continue;
		}
		if (p->node->MQTT_Unpublish.SubToPublish_flag == 1)
		{
			return p->node;
		}
		p = tmp;
	}
	return NULL;
}
/// <summary>
/// 通过发布消息主题寻找与订阅主题相符的客户端
/// </summary>
/// <param name="publishtpoic">接收到的publish主题</param>
/// <returns>与订阅主题相符的客户端句柄</returns>
MQTT* List_node_by_PublishTpoic(char* publishtpoic)
{
	MQTT_List* tmp, * p = root;
	int len = 0;
	if (publishtpoic == NULL)
		return NULL;
	len = List_Length(root);
	while (len-- && p != NULL)
	{
		tmp = p->prev_node;
		if (p->node == NULL)
		{
			p = tmp;
			continue;
		}
		if (memcmp(publishtpoic, p->node->MQTT_Subscribe.SubscribeTpioc, strlen(publishtpoic)) == 0)
		{
			return p->node;
		}
		p = tmp;
	}
	return NULL;
}
void List_type_print_info(List_type* List_node)
{
	DEBUG_STRING("--------------------------------------------------------------------", " ");
	if (List_node == NULL)
		return;
	DEBUG_NUMBER("链表元素个数", List_Length(root));
	DEBUG_STRING("IP", List_node->IP);
	DEBUG_NUMBER("Port", List_node->port);
	DEBUG_NUMBER("Current_State", List_node->Current_State);
	DEBUG_NUMBER("mqtt_socket_type", List_node->mqtt_socket_type);

	DEBUG_STRING("ClientID", List_node->ClientID);
	DEBUG_STRING("UserName", List_node->UserName);
	DEBUG_STRING("PassWord", List_node->PassWord);

	DEBUG_NUMBER("Keep_Alive", List_node->MQTT_Connect.Keep_Alive);
	DEBUG_STRING("Will_Topic", List_node->MQTT_Connect.Will_Topic);
	DEBUG_STRING("Will_Message", List_node->MQTT_Connect.Will_Message);
	DEBUG_NUMBER("connectflags", List_node->MQTT_Connect.connectflags.Flags);
	DEBUG_NUMBER("currentCleanSession", List_node->MQTT_Connect.currentCleanSession);
	DEBUG_NUMBER("Connack_ErrCode", List_node->MQTT_Connect.Connack_ErrCode);
	DEBUG_NUMBER("connect_state", List_node->MQTT_Connect.connect_state);

	DEBUG_STRING("PublishTpoic", List_node->MQTT_Publish.PublishTpioc);
	DEBUG_NUMBER("Qos", List_node->MQTT_Publish.Qos);
	DEBUG_NUMBER("Packet_Identifier", List_node->MQTT_Publish.Packet_Identifier);
	DEBUG_NUMBER("RETAIN", List_node->MQTT_Publish.RETAIN);
	if (List_node->MQTT_Publish.message != NULL)
	{
		DEBUG_STRING("message", List_node->MQTT_Publish.message);
		DEBUG_NUMBER("Dup", List_node->MQTT_Publish.Dup);
	}

	DEBUG_STRING("SubscribeTpioc", List_node->MQTT_Subscribe.SubscribeTpioc);
	DEBUG_NUMBER("Subscribe_Result", List_node->MQTT_Subscribe.Subscribe_Result);
	DEBUG_NUMBER("SubscribeQos", List_node->MQTT_Subscribe.Qos);
	DEBUG_NUMBER("SUBSCRIBE_Identifier", List_node->MQTT_Subscribe.SUBSCRIBE_Identifier);
	DEBUG_STRING("--------------------------------------------------------------------", " ");

}
void List_print(MQTT_List* Item)
{
	MQTT_List* tmp;
	int len = 0;
	if (Item == NULL)
		return;
	len = List_Length(Item);
	while (len-- && Item != NULL)
	{
		tmp = Item->prev_node;
		if (Item->node == NULL)
		{
			Item = tmp;
			continue;
		}
		List_type_print_info(Item->node);
		Item = tmp;
	}
}


